package com.tender.integration.websocket.service.impl;

import com.alibaba.fastjson.JSON;
import com.tender.condition.BizMsgAssoUserCondition;
import com.tender.constants.Constants;
import com.tender.constants.WebSocketConstants;
import com.tender.entity.BizMsgAssoUserEntity;
import com.tender.enums.ErrorTypeEnum;
import com.tender.enums.MsgHasReadEnum;
import com.tender.exceptions.WebSocketException;
import com.tender.integration.config.properties.MsgProperties;
import com.tender.integration.websocket.pojo.WebSocketWrapperInfo;
import com.tender.integration.websocket.service.WebSocketService;
import com.tender.integration.websocket.converter.WebSocketConverter;
import com.tender.integration.websocket.utils.WebSocketMapUtil;
import com.tender.service.BizMsgAssoUserService;
import com.tender.utils.LoginUser;
import com.tender.utils.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.CollectionUtils;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * 【消息中心】：直接查【消息中心分页接口】，【已读】、【未读】则利用用户消息表来实现（用户已读消息，则user_id 和 msg_id 绑定入库）
 * 【审核通过消息】：则利用 WebSocket 来推送
 * 1、审核通过后，如果用户在线，则直接推送消息给前端
 * 2、如果用户离线，则消息入待发送表，（建立连接的时候，用户状态存入redis中，断开连接的时候，清除用户状态）
 * 2.1、当有新用户连线，就尝试发送消息到该用户，利用 WebSocket 集群特性。
 * 2.2、定时扫描历史信息，来尝试发送给用户。（防止一直没人登录）
 */
@Slf4j
public class WebSocketServiceImpl implements WebSocketService {

    @Autowired
    private BizMsgAssoUserService bizMsgAssoUserService;
    @Autowired
    private RedisUtils redisUtils;
    @Autowired
    private MsgProperties msgProperties;
    @Autowired
    private Environment environment;
    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    private final String myApplicationName = System.getProperty(Constants.APPLICATION_NAME);

    @Override
    public void handleOpen(WebSocketSession session) {
        // 这个时候就需要在建立 webSocket 时存储的 用户信息了
        Map<String, Object> attributes = session.getAttributes();
        LoginUser userInfo = (LoginUser) attributes.get(WebSocketConstants.WS_USER_INFO_KEY);
        // 用户 id 和 session 关联起来
        // ConcurrentWebSocketSessionDecorator 线程安全的
        Map<String, WebSocketWrapperInfo> webSocketInfoMap = WebSocketMapUtil.getByUserId(String.valueOf(userInfo.getUserId()));
        if (null != webSocketInfoMap) {
            WebSocketWrapperInfo webSocketWrapperInfo = webSocketInfoMap.get(userInfo.getDevice());
            if (null != webSocketWrapperInfo) {
                // 有重复连接，需要关闭再重新连接
                try {
                    webSocketWrapperInfo.getWebSocketSession().close();
                } catch (IOException e) {
                    // TODO 关闭异常 抛异常?
                    throw new WebSocketException(ErrorTypeEnum.WEB_SOCKET_CLOSE_EXP);
                }
            }
        }

        // 包装 WebSocketSession【WebSocketWrapperInfo】，然后保存到 内存 Map 中
        String port = environment.getProperty(Constants.SERVER_PORT);
        WebSocketMapUtil.putMap(String.valueOf(userInfo.getUserId()), WebSocketConverter.prepareWebSocketInfo(port, session, userInfo), userInfo.getDevice());

        // 记录用户登录状态到 redis 中
        String userWsCntStatusKey = String.format(WebSocketConstants.WS_USER_CNT_STATUS_KEY, userInfo.getUserId());
        // TODO 用户web socket 连接状态没有设置过期时间
        // 同一个用户，对应多个客户端 可连接到 web socket server
        redisUtils.setHash(
                userWsCntStatusKey,
                userInfo.getDevice(),
                true   // redis value
        );

        // 每个用户需要与唯一应用名绑定，同一个用户，不同设备，不影响。
        redisUtils.set(String.valueOf(userInfo.getUserId()),
                myApplicationName); // 这是 web socket server 集群中独一无二的应用名称

        // 异步推送消息
        // 扫描业务消息表，获取该用户相关的业务消息，然后进行推送
        asyncProcessUserUnreadMsg(userInfo);

        log.info("a new connection opened，current online count：{}", WebSocketMapUtil.getUserNum());
    }


    @Override
    public void handleClose(WebSocketSession session) {
        Map<String, Object> attributes = session.getAttributes();
        LoginUser userInfo = (LoginUser) attributes.get(WebSocketConstants.WS_USER_INFO_KEY);
        WebSocketMapUtil.deleteByUserIdAndDevice(String.valueOf(userInfo.getUserId()), userInfo.getDevice());
        // redis 清除用户登录状态信息
        String userWsCntStatusKey = String.format(WebSocketConstants.WS_USER_CNT_STATUS_KEY, userInfo.getUserId());
        Map<Object, Object> map = redisUtils.getHash(userWsCntStatusKey);
        if (null != map) {
            if (0 == map.size()) {
                redisUtils.delete(userWsCntStatusKey);
                // 删除 用户 和 应用名称 的绑定关系
                redisUtils.delete(String.valueOf(userInfo.getUserId()));
            }
            if (map.containsKey(userInfo.getDevice())) {
                redisUtils.deleteHash(userWsCntStatusKey, userInfo.getDevice());
                map.remove(userInfo.getDevice());
            }
            if (0 == map.size()) {
                redisUtils.delete(userWsCntStatusKey);
                // 用户和应用名称 绑定关系处理
                // 删除 用户 和 应用名称 的绑定关系
                redisUtils.delete(String.valueOf(userInfo.getUserId()));
            }
        }

        // 断开 session
        try {
            session.close();
        } catch (IOException e) {
            // TODO
            // do nothing
        }

        log.info("a new connection closed，current online count：{}", WebSocketMapUtil.getUserNum());
    }

    @Override
    public void handleMessage(WebSocketSession session, String message) {
        this.handleMessage(session, new TextMessage(message));
    }

    /**
     * 所有的方法，最后都会调用这个 handleMessage 方法，来处理消息
     */
    @Override
    public void handleMessage(WebSocketSession session, TextMessage message) {
        // 处理前端传来的文本消息
        Map<String, Object> attributes = session.getAttributes();
        LoginUser userInfo = (LoginUser) attributes.get(WebSocketConstants.WS_USER_INFO_KEY);
        // redis 中用户状态续命
        if (session.isOpen()) {
            try {
                session.sendMessage(message);
            } catch (IOException e) {
                // TODO
                // do nothing!
            }
        }
        log.info("received a message：{}", message);
    }

    @Override
    public void sendMessage(String userId, String device, String message) throws IOException {
        this.sendMessage(userId, device, new TextMessage(message));
    }

    @Override
    public void sendMessage(String userId, String device, TextMessage textMessage) throws IOException {
        WebSocketWrapperInfo webSocketWrapperInfo = WebSocketMapUtil.getByUserIdAndDevice(userId, device);
        if (null != webSocketWrapperInfo && null != webSocketWrapperInfo.getWebSocketSession()) {
            if (webSocketWrapperInfo.getWebSocketSession().isOpen()) {
                this.sendMessage(webSocketWrapperInfo.getWebSocketSession(), textMessage);
            }
        }
    }

    @Override
    public void sendMessage(WebSocketSession session, String message) throws IOException {
        this.sendMessage(session, new TextMessage(message));
    }

    @Override
    public void sendMessage(WebSocketSession session, TextMessage textMessage) throws IOException {
        this.handleMessage(session, textMessage);
    }

    @Override
    public void sendMessage(String userId, String message) throws IOException {
        this.sendMessage(userId, new TextMessage(message));
    }

    @Override
    public void sendMessage(String userId, TextMessage message) throws IOException {
        Map<String, WebSocketWrapperInfo> device2WsMap = WebSocketMapUtil.getByUserId(userId);
        if(!CollectionUtils.isEmpty(device2WsMap)) {
            device2WsMap.forEach((device, webSocketWrapperInfo) -> {
                if (null != webSocketWrapperInfo && null != webSocketWrapperInfo.getWebSocketSession()) {
                    if (webSocketWrapperInfo.getWebSocketSession().isOpen()) {
                        try {
                            this.sendMessage(webSocketWrapperInfo.getWebSocketSession(), message);
                        } catch (IOException e) {
                            // TODO
                            // do nothing!
                        }
                    }
                }
            });
        }
    }

    @Override
    public void broadCast(String message) throws IOException {
        this.broadCast(new TextMessage(message));
    }

    @Override
    public void broadCast(TextMessage textMessage) throws IOException {
        // 每个用户对应的所有设备都广播消息
        WebSocketMapUtil.values().forEach(webSocketWrapperInfoMap -> {
            webSocketWrapperInfoMap.forEach((device, webSocketWrapperInfo) -> {
                if (webSocketWrapperInfo.getWebSocketSession().isOpen()) {
                    try {
                        this.sendMessage(webSocketWrapperInfo.getWebSocketSession(), textMessage);
                    } catch (IOException e) {
                        // TODO 异常自定义
                        //. 消息需要吞掉、发送失败的消息落DB，后续再补偿？
                    }
                }
            });
        });
    }

    @Override
    public void handleError(WebSocketSession session, Throwable error) {
        handleClose(session);
        log.error("websocket error：{}，session id：{}", error.getMessage(), session.getId());
        log.error("specific error: ", error);
    }

    private void asyncProcessUserUnreadMsg(LoginUser userInfo) {
        try {
            threadPoolTaskExecutor.execute(() -> {
                // 用户每次登录，需要查询用户对应的未读消息，并进行推送
                BizMsgAssoUserCondition condition = new BizMsgAssoUserCondition();
                condition.setUserId(userInfo.getUserId());
                condition.setHasRead(MsgHasReadEnum.NOT_READ.getCode());
                List<BizMsgAssoUserEntity> bizMsgRecordDTOS = bizMsgAssoUserService.selectList(condition);
                if (!CollectionUtils.isEmpty(bizMsgRecordDTOS)) {
                    log.info("【推送消息】userId: {}, 待推送消息: {}", userInfo.getUserId(), JSON.toJSONString(bizMsgRecordDTOS));
                    try {
                        // TODO 每个设备都推送 ?
                        this.sendMessage(String.valueOf(userInfo.getUserId()), JSON.toJSONString(bizMsgRecordDTOS));
                    } catch (IOException e) {
                        log.info("***********************************************");
                        log.error("【推送消息】消息列表：{}, 异常：", JSON.toJSONString(bizMsgRecordDTOS), e);
                        log.info("***********************************************");
                    }
                }
            });
        } catch (Exception e) {
            log.error("【推送消息】提交待处理的消息列表任务异常：", e);
        }
    }

}
